/*
 * SmartCam master processor library
 * Miscellaneous management
 */

#define _XOPEN_SOURCE 500

#include <string.h>
#include <getopt.h>
#include <stdarg.h>
#include <log.h>

#include <scc.h>
#include <scu.h>
#include <scilp.h>
#ifdef PSOS
#include <scxtl.h>
#else
#include <scsim.h>
#endif

#include "globals.h"
#include "results.h"
#include "management.h"
#include "settings.h"

table_t sc_procs;
table_t sc_ops;
table_t sc_vars;
table_t sc_streams;
table_t sc_bufs;
table_t sc_temps;
table_t sc_origins;
table_t sc_threads;

sc_op_t source;

static scos_mutex_t sc_lock;

void scManReadMachines(char *file);

void *schedthr(void *dummy)
{
  while (1)
  {
    lprintf(STREAM(scl.crawl), "SCHED");
    sched_yield();
  }

  return NULL;
}

void scManInit(void)
{
//  pthread_t thread;

  lprintf(STREAM(scl.notice), "Initializing management");

//  scos_thread_create(&thread, schedthr, NULL);

  abortOnNULL(STREAM(scl.err), settings.name);
  abortOnNULL(STREAM(scl.err), settings.machinesfile);

  if (settings.writetodisk)
    scuBufferSetGlobalOption(SCU_BUF_OPTION_WRITETODISK, settings.writetodisk);

  abortOnNegative(STREAM(scl.err), tblcreate(&sc_procs, sizeof(sc_proc_t)));
  abortOnNegative(STREAM(scl.err), tblcreate(&sc_ops, sizeof(sc_op_t)));
  abortOnNegative(STREAM(scl.err), tblcreate(&sc_vars, sizeof(sc_var_t)));
  abortOnNegative(STREAM(scl.err), tblcreatehash(&sc_vars, sizeof(void*), 64));
  abortOnNegative(STREAM(scl.err), tblcreate(&sc_temps, sizeof(sc_temp_t)));
  abortOnNegative(STREAM(scl.err), tblcreatehash(&sc_temps, sizeof(void*), 8));
  abortOnNegative(STREAM(scl.err), tblcreate(&sc_bufs, sizeof(sc_buf_t)));
  abortOnNegative(STREAM(scl.err), tblcreate(&sc_origins, sizeof(sc_origin_t)));
  abortOnNegative(STREAM(scl.err), tblcreate(&sc_threads, sizeof(sc_thread_t)));
  abortOnNegative(STREAM(scl.err), tblcreatehash(&sc_threads, sizeof(int), 31));

  scManAddThread();

  if (settings.machinesfile)
    scManReadMachines(settings.machinesfile);
}

void scManReadMachines(char *file)
{
  int id=-1, myaddrlen=0;
  sc_proc_t *p;
  struct sockaddr myaddr;

  scs_target_t *t;

  lprintf(STREAM(scl.info), "Reading machine configuration from %s", file);

  scSetRead(file);
  scSetIndex();

  memset(&myaddr, 0, sizeof(myaddr));
  while (tblgetnextelement(&scs_targets, &id, &t))
  {
    /* Targets without an address are superclasses; not real. */
    if (t->address && (t->routing != 1 || settings.simulate))
    {      
      abortOnNegative(STREAM(scl.err), tbladdnewi(&sc_procs, &p, id));
      p->id = id;
      abortOnNULL(STREAM(scl.err), p->name = (char*)malloc(strlen(t->name)+1));
      strcpy(p->name, t->name);

      /* Interpret address */
      if (t->address[0] == '/')
      {
	p->address.sa_family = AF_UNIX;
	p->addrlen = strlen(t->address) + sizeof(p->address.sa_family);
	strcpy(p->address.sa_data, t->address);
      }
      else
      {
	p->address.sa_family = AF_INET;
	p->addrlen = sizeof(p->address);
#ifndef PSOS
	abortOnNULL(STREAM(scl.err), ((struct sockaddr_in*)(&p->address))->sin_addr.s_addr = inet_addr(t->address));
#endif	
      }

      if (!strcmp(p->name, settings.name))
      {
	memcpy(&myaddr, &p->address, sizeof(myaddr));
	myaddrlen = p->addrlen;
      }

      /* Interpret driver */
      p->type = -1;
      if (t->driver)
      {
	if (!strcmp(t->driver, "ilp"))
	  p->type = SC_PROC_ILP;
        else if (!strcmp(t->driver, "xtl"))
	  p->type = SC_PROC_XTL;
      }

      abortOnNegative(STREAM(scl.err), tblcreate(&p->operations, sizeof(sc_op_t)));
      abortOnNegative(STREAM(scl.err), tblcreate(&p->buffers, sizeof(sc_buf_t)));

      if (t->buffer > 0)
	p->bufsize = t->buffer;
      else
	p->bufsize = SC_DEFAULT_BUFFER;
    }
  }

  sccInit(&myaddr, myaddrlen);

  if (settings.master)
    sccMasterSetup(scResCallback, NULL);

  id = -1;
  while (tblget(&sc_procs, id = tblgetnexti(&sc_procs, id), &p))
  {
    if (settings.simulate)
    {
      scSimInit(p);
    }
    else
    {
      if (!memcmp(&myaddr, &p->address, sizeof(myaddr)))
      {
	switch (p->type)
	{
	  case SC_PROC_ILP:
	    scIlpInit(p->id);
	    break;
#ifdef PSOS
	  case SC_PROC_XTL:
	    scXtlInit(p->id);
	    break;
#endif
	  default:
	    lprintf(STREAM(scl.err), "Unknown processor type %d for processor '%s'", p->type, p->name);
	}
      }
    }
  }

  scos_mutex_init(&sc_lock);
}

#if 1
table_t scManArgDecode(char *format, va_list ap)
{
  int c;
  table_t args;
  sc_argdecode_t *d;
  
  abortOnNegative(STREAM(scl.err), tblcreate(&args, sizeof(sc_argdecode_t)));

  for (c=0; c < strlen(format); c++)
  {
    abortOnNegative(STREAM(scl.err), tbladdnew(&args, &d));
    
    switch (format[c])
    {
      case 'i':
	d->type = SC_VAR_STREAM;
	d->intent = SC_INTENT_IN;
	d->address = va_arg(ap, STREAM*);
 	d->size = sizeof(int); /* Well, 0, really, but malloc(0) may behave strangely */
	break;
      case 'I':
	d->type = SC_VAR_STREAM;
	d->intent = SC_INTENT_OUT;
	d->address = va_arg(ap, STREAM*);
	d->size = sizeof(int); /* Well, 0, really, but malloc(0) may behave strangely */
	break;
      case 'd':
	d->type = SC_VAR_INTEGER;
	d->intent = SC_INTENT_IN;
	d->address = va_arg(ap, int*);
	d->size = sizeof(int);
	break;
      case 'D':
	d->type = SC_VAR_INTEGER;
	d->intent = SC_INTENT_OUT;
	d->address = va_arg(ap, int*);
	d->size = sizeof(int);
	break;
      case 'f':
	d->type = SC_VAR_DOUBLE;
	d->intent = SC_INTENT_IN;
	d->address = va_arg(ap, double*);
	d->size = sizeof(double);
	break;
      case 'F':
	d->type = SC_VAR_DOUBLE;
	d->intent = SC_INTENT_OUT;
	d->address = va_arg(ap, double*);
	d->size = sizeof(double);
	break;    
      default:
	abortErrorl((STREAM(scl.err), "Unknown argument type %c for arg %d", format[c], c));
    }
  }

  return args;
}
#else
table_t scManArgDecode(char *format, int op, va_list ap)
{
  int c;
  table_t args;
  sc_argdecode_t *d;
  
  abortOnNegative(STREAM(scl.err), tblcreate(&args, sizeof(sc_argdecode_t)));

  for (c=0; c < strlen(format); c++)
  {
    abortOnNegative(STREAM(scl.err), tbladdnew(&args, &d));
    
    switch (format[c])
    {
      case 'i':
	d->type = SC_VAR_STREAM;
	d->intent = SC_INTENT_IN;
	d->address = va_arg(ap, STREAM*);
 	d->size = sizeof(int); /* Well, 0, really, but malloc(0) may behave strangely */
	break;
      case 'I':
	d->type = SC_VAR_STREAM;
	d->intent = SC_INTENT_OUT;
	d->address = va_arg(ap, STREAM*);
	d->size = sizeof(int); /* Well, 0, really, but malloc(0) may behave strangely */
	break;
      case 'd':
	d->type = SC_VAR_INTEGER;
	d->intent = SC_INTENT_IN;
	d->address = va_arg(ap, int*);
	d->size = sizeof(int);
	break;
      case 'D':
	d->type = SC_VAR_INTEGER;
	d->intent = SC_INTENT_OUT;
	d->address = va_arg(ap, int*);
	d->size = sizeof(int);
	break;
      case 'f':
	d->type = SC_VAR_DOUBLE;
	d->intent = SC_INTENT_IN;
	d->address = va_arg(ap, double*);
	d->size = sizeof(double);
	break;
      case 'F':
	d->type = SC_VAR_DOUBLE;
	d->intent = SC_INTENT_OUT;
	d->address = va_arg(ap, double*);
	d->size = sizeof(double);
	break;    
      default:
	abortErrorl((STREAM(scl.err), "Unknown argument type %c for arg %d", format[c], c));
    }
  }

  return args;
}
#endif

int scManIsVariableFinalized(sc_var_t *v)
{
  int id;
  sc_instance_t *i;

  id = -1;
  while (tblget(&v->instances, id = tblgetnexti(&v->instances, id), &i))
    if (!i->finalized) return 0;

  return 1;
}

int scManIsVariableDispatched(sc_var_t *v)
{
  int id;
  sc_arg_t *a;

  if (!scManIsVariableFinalized(v)) return 0;

  if (v->producer && !v->producer->dispatched) return 0;

  id = -1;
  while (tblget(&v->consumers, id = tblgetnexti(&v->consumers, id), &a))
    if (!a->dispatched) return 0;

  return 1;
}

/* VARIABLE MUST BE LOCKED */
int scManIsVariableDestroyed(sc_var_t *v)
{
  int id;
  sc_buf_t *b;

  abortOnError(STREAM(scl.err), v->type != SC_VAR_STREAM);

  if (!scManIsVariableDispatched(v)) return 0;

  id = -1;
  while (tblget(&v->buffers, id = tblgetnexti(&v->buffers, id), &b))
    if (!b->destroyed) return 0;

  return 1;
}

sc_instance_t *scManFindInstance(void *address)
{
  sc_var_t *v;
  sc_instance_t *i;

  if (!tblget(&sc_vars, tblfindhash(&sc_vars, &address), &v)) return NULL;
  if (!tblget(&v->instances, tblfindhash(&v->instances, &address), &i)) return NULL;

  return i;
}

sc_var_t *scManAddVariable(int type, int size)
{
  int id;
  sc_var_t *v;  

  abortOnNegative(STREAM(scl.err), id = tbladdnew(&sc_vars, &v));

  v->id = id;
  v->type = type;
  v->size = size;

  if (type == SC_VAR_STREAM)
  {
    abortOnNegative(STREAM(scl.err), tblcreate(&v->buffers, sizeof(sc_buf_t)));
    v->valid = 1;
  }

  abortOnNegative(STREAM(scl.err), tblcreate(&v->consumers, sizeof(sc_arg_t)));
  abortOnNegative(STREAM(scl.err), tblcreate(&v->instances, sizeof(sc_instance_t)));  
  abortOnNegative(STREAM(scl.err), tblcreatehash(&v->instances, sizeof(void*), 8));
  abortOnNegative(STREAM(scl.err), tblcreate(&v->originrefs, sizeof(sc_originref_t)));  
  abortOnNegative(STREAM(scl.err), tblcreate(&v->transports, sizeof(sc_transport_t)));

  scos_mutex_init(&v->lock);
  scos_cond_init(&v->condition);

  return v;
}

sc_instance_t *scManAddInstance(sc_var_t *v, void *address)
{
  int id;
  sc_instance_t *i;

  abortOnNegative(STREAM(scl.err), id = tbladdnew(&v->instances, &i));

  i->id = id;
  i->variable = v;
  i->address = address;
  if (v->type != SC_VAR_STREAM || settings.multiuse) i->multiuse = 1;

  if (address)
  {
    abortOnNegative(STREAM(scl.err), tbladdhash(&sc_vars, &address, v->id));
    abortOnNegative(STREAM(scl.err), tbladdhash(&v->instances, &address, i->id));

    lprintf(STREAM(scl.crawl), "Added variable instance %d.%d with address %p",
	    v->id, i->id, i->address);
  }
  else
  {
    lprintf(STREAM(scl.crawl), "Added unreferenceable variable instance %d.%d",
	    v->id, i->id);
  }

  return i;
}

sc_buf_t *scManAddBuffer(sc_var_t *v, sc_proc_t *p)
{
  sc_buf_t *b;
  int id;

  abortOnNegative(STREAM(scl.err), id=tbladdnew(&sc_bufs, &b));
  abortOnNegative(STREAM(scl.err), tbladdi(&v->buffers, b, p->id));
  abortOnNegative(STREAM(scl.err), tbladd(&p->buffers, b));
  
  b->id = id;
  b->stream = v;
  b->processor = p;
  b->size = p->bufsize;

  return b;
}

void *scManAddTemporary(int size)
{
  sc_temp_t *t;
  int id;

  abortOnNegative(STREAM(scl.err), id = tbladdnew(&sc_temps, &t));

  t->id = id;
  t->address = malloc(size);

  abortOnNegative(STREAM(scl.err), tbladdhash(&sc_temps, &t->address, t->id));

  lprintf(STREAM(scl.crawl), "Added temporary %d with address %p",
	  t->id, t->address);

  return t->address;  
}

void scManCheckDestroyTemporary(void *address)
{
  sc_temp_t *t;

  if (!tblget(&sc_temps, tblfindhash(&sc_temps, &address), &t)) return;

  lprintf(STREAM(scl.crawl), "Removing temporary %d", t->id);

  free(t->address);
  abortOnError(STREAM(scl.err), tbldelhash(&sc_temps, &address));
  abortOnNegative(STREAM(scl.err), tbldel(&sc_temps, t->id, 1));
}

void scManCreateOrigin(sc_var_t *v, int isVirtual)
{
  sc_origin_t *o;
  sc_originref_t *r;
  int id;

  abortOnNegative(STREAM(scl.err), id = tbladdnew(&sc_origins, &o));

  o->id = id;
  o->isVirtual = isVirtual;
  abortOnError(STREAM(scl.err), tblcreate(&o->references, sizeof(sc_originref_t)));
  abortOnNegative(STREAM(scl.err), id = tbladdnew(&o->references, &r));
  r->id = id;

  abortOnNegative(STREAM(scl.err), id = tbladd(&v->originrefs, r));
  r->origin = o;
  r->variable = v;
  r->delay = 0;

  lprintf(STREAM(scl.crawl), "Added origin %d, with seed %d", o->id, v->id);
}

void scManCheckReferenceOrigin(sc_var_t *v, sc_originref_t *r, int delay)
{
  int id, id2, id3;
  sc_originref_t *r2;
  sc_originref_t *cr1, *cr2, *cr3;

  /* *** NEED TO FIND OUT IF THIS IS NOT TOO SLOW *** */

  /* Check if we already reference this origin */
  id = -1;
  while (tblgetnextelement(&v->originrefs, &id, &cr1))
    if (cr1->origin->id == r->origin->id)
    {
      cr1->delay = MAX(cr1->delay, r->delay + delay);
      return;
    }

  /* Check if there's already an origin v shares with all
   * r->origin->references->variable's */
  id = -1;
  while (tblgetnextelement(&v->originrefs, &id, &cr1))
  {
    id2 = -1;
    while (tblgetnextelement(&r->origin->references, &id2, &cr2))
    {
      id3 = -1;
      while (tblgetnextelement(&cr2->variable->originrefs, &id3, &cr3))
      {
	if (cr1->origin->id == cr3->origin->id)
	  /* We share cr1->origin with cr2->variable */
	  break;
      }
      
      if (id3 == -1)
	/* We don't share cr1->origin with cr2->variable */
	break;
    }
    
    if (id2 == -1)
      /* We share cr1->origin with all of r->origin->references->variable's */
      return;
  }
  
  abortOnNegative(STREAM(scl.err), id = tbladdnew(&r->origin->references, &r2));  
  r2->id = id;
  abortOnNegative(STREAM(scl.err), id = tbladd(&v->originrefs, r2));

  r2->origin = r->origin;
  r2->variable = v;
  r2->delay = r->delay + delay;

  lprintf(STREAM(scl.crawl), "Added reference for %d to origin %d with delay %d (from %d)", v->id, r->origin->id, r2->delay, r->variable->id);
}

void scManReplaceOrigin(sc_origin_t *o, table_t *references)
{
  int id=-1, id2;
  sc_originref_t *r, *r2;
  sc_var_t *v;

  /* Loop over the virtual origin's references */
  while (tblgetnextelement(&o->references, &id, &r))
  {
    v = r->variable;

    /* Find reference to virtual origin in the variable's table */
    id2 = -1;
    while (tblgetnextelement(&v->originrefs, &id2, &r2))
    {    
      if (r2->origin == o)
      {
	/* Delete it */
	abortOnNegative(STREAM(scl.err), tbldel(&v->originrefs, id2, 0));
	break;
      }
    }

    /* Add references to the actual origins to the variable's table */
    id2 = -1;
    while (tblgetnextelement(references, &id2, &r2))
    {
      scManCheckReferenceOrigin(v, r2, r->delay);
    }
  }

  abortOnError(STREAM(scl.err), tblclear(&o->references, 1));
  scManCheckDestroyOrigin(o);
}

void scManCheckDestroyOrigin(sc_origin_t *o)
{
  if (tblgetelements(&o->references)) return;

  lprintf(STREAM(scl.crawl), "Removing origin %d", o->id);

  abortOnError(STREAM(scl.err), tbldestroy(&o->references, 1));
  abortOnNegative(STREAM(scl.err), tbldel(&sc_origins, o->id, 1));
}

void scManAcquireLock(void)
{
#if 0
  sc_thread_t *t = scManGetThread();
#endif

  scos_mutex_lock(&sc_lock);

#if 0
  if (t)
  {
    t->locks++;
    lprintf(STREAM(scl.err), "Thread %d acquired global lock (%d)", t->id, t->locks);
  }
  else
    lprintf(STREAM(scl.err), "Unknown thread acquired global lock");
#endif
}

void scManShedLock(void)
{
#if 0
  sc_thread_t *t = scManGetThread();
#endif

  scos_mutex_unlock(&sc_lock);

#if 0
  if (t)
  {
    t->locks--;
    lprintf(STREAM(scl.err), "Thread %d shed global lock (%d)", t->id, t->locks);
  }
  else
    lprintf(STREAM(scl.err), "Unknown thread shed global lock");
#endif
}

void scManCheckDestroyArgument(sc_arg_t *a)
{
  int id=-1;
  sc_op_t *o = a->operation;
  sc_instance_t *i = a->instance;
  sc_arg_t *a2;

  if (settings.tracefile) return;

  /* PRECONDITION: Argument has been dispatched (or received for outputs) */
  if (!a->dispatched) return;

  /* PRECONDITION: If producer, variable is finalized and dispatched */
  if (a->instance->variable->producer == a)
  {
    if (!scManIsVariableFinalized(a->instance->variable)) return;
    if (!scManIsVariableDispatched(a->instance->variable)) return;
  }

  lprintf(STREAM(scl.crawl),
	  "Destroying %s(%d) arg %d (instance %d.%d)",
	  a->operation->text, a->operation->id, a->id, a->instance->variable->id,
	  a->instance->id);

  /* Remove ourselves from variable */
  if (a->intent == SC_INTENT_OUT)
  {
    a->instance->variable->producer = NULL;
  }
  else
  {
    while (tblget(&a->instance->variable->consumers,
		  id = tblgetnexti(&a->instance->variable->consumers, id),
		  &a2))
      if (a == a2)
	abortOnNegative(STREAM(scl.err), id = tbldel(&a->instance->variable->consumers, id, 0));
  }

  /* And from operation */
  abortOnNegative(STREAM(scl.err), tbldel(&a->operation->arguments, a->id, 1));
  
  /* Now.... Since the producer cannot be destroyed until all consumers are dispatched,
   * do it here. A caveat is that the producer might destroy our own instance, if it
   * is occupying the same one. Thus, only try to destroy that when it is not.
   */
  if (i->variable->producer)
  {
    if (i->variable->producer->instance == i)
    {
      scManCheckDestroyArgument(i->variable->producer);
    }
    else
    {
      scManCheckDestroyArgument(i->variable->producer);
      scManCheckDestroyInstance(i);
    }
  }
  else
    scManCheckDestroyInstance(i);

  scManCheckDestroyOperation(o);
}

void scManCheckDestroyInstance(sc_instance_t *i)
{
  int id=-1;
  sc_arg_t *a;
  sc_var_t *v = i->variable;
  
  if (settings.tracefile) return;

  /* PRECONDITION: Instance has been finalized */
  if (!i->finalized) return;

  /* PRECONDITION: No one is blocking on the instance */
  if (i->blocked) return;

  /* PRECONDITION: No arguments reference the instance */
  if (i->variable->producer && (i->variable->producer->instance == i)) return;
  while (tblget(&i->variable->consumers, id = tblgetnexti(&i->variable->consumers, id), &a))
    if (a->instance == i) return;

  lprintf(STREAM(scl.crawl), "Destroying instance %d.%d", i->variable->id, i->id);
  
  scos_mutex_lock(&v->lock);

  /* Remove ourselves from the variable */
  abortOnNegative(STREAM(scl.err), tbldel(&i->variable->instances, i->id, 1));

  scos_mutex_unlock(&v->lock);
  
  scManCheckDestroyVariable(v);
}

void scManCheckDestroyVariable(sc_var_t *v)
{
  int id=-1;
  sc_origin_t *o;
  sc_originref_t *r;

  if (settings.tracefile) return;

  /* PRECONDITION: Not protected (ugly) */
  if (v->pentagram) return;

  /* PRECONDITION: No instances reference the variable */
  if (tblgetelements(&v->instances)) return;

  /* PRECONDITION: No arguments reference the variable */
  if (v->producer) return;
  if (tblgetelements(&v->consumers)) return;

  /* PRECONDITION: No buffers reference the variable */
  if (v->type == SC_VAR_STREAM && tblgetelements(&v->buffers)) return;

  lprintf(STREAM(scl.crawl), "Destroying variable %d", v->id);

  /* Free data */
  if (v->data) free(v->data);

  /* If we were a stream, free buffer table (empty) */
  if (v->type == SC_VAR_STREAM)
    abortOnNegative(STREAM(scl.err), tbldestroy(&v->buffers, 0));

  /* Free consumer and instances tables (emtpy) */
  abortOnNegative(STREAM(scl.err), tbldestroy(&v->consumers, 0));
  abortOnNegative(STREAM(scl.err), tbldestroyhash(&v->instances));
  abortOnNegative(STREAM(scl.err), tbldestroy(&v->instances, 0));

  /* Remove origin references */
  while (tblgetnextelement(&v->originrefs, &id, &r))
  {
    o = r->origin;
    abortOnNegative(STREAM(scl.err), tbldel(&r->origin->references, r->id, 1));
    scManCheckDestroyOrigin(o);
  }

  abortOnNegative(STREAM(scl.err), tbldestroy(&v->originrefs, 0));

  /* And destroy locking mechanisms */
  scos_mutex_destroy(&v->lock);
  scos_cond_destroy(&v->condition);
  
  /* Remove ourselves from global variable list */
  abortOnNegative(STREAM(scl.err), tbldel(&sc_vars, v->id, 1));
}

void scManCheckDestroyOperation(sc_op_t *o)
{
  int id=-1;
  sc_op_t *o2;

  if (settings.tracefile) return;

  /* PRECONDITION: Operation has finished */
  if (!o->finished) return;

  /* PRECONDITION: No arguments reference the operation */
  if (tblgetelements(&o->arguments)) return;

  lprintf(STREAM(scl.crawl), "Destroying %s(%d)", o->text, o->id);
  
  /* Destroy argument table (empty) */
  abortOnNegative(STREAM(scl.err), tbldestroy(&o->arguments, 0));

  /* Remove from processor op list */
  while (tblget(&o->processor->operations, 
		id = tblgetnexti(&o->processor->operations, id), &o2))
    if (o == o2)
      abortOnNegative(STREAM(scl.err), id = tbldel(&o->processor->operations, id, 0));

  /* ... and global operation list */
  abortOnNegative(STREAM(scl.err), tbldel(&sc_ops, o->id, 1));
}

void scManCheckDestroyBuffer(sc_buf_t *b)
{
  int id=-1;
  sc_buf_t *b2;
  sc_var_t *v = b->stream;

  if (settings.tracefile) return;

  /* PRECONDITION: Buffer has been destroyed */
  if (!b->destroyed) return;

  lprintf(STREAM(scl.crawl), "Destroying buffer %d (%s)", b->id, b->processor->name);

  scos_mutex_lock(&v->lock);

  /* Remove ourselves from variable */
  while (tblget(&b->stream->buffers, id = tblgetnexti(&b->stream->buffers, id), &b2))
    if (b == b2)
      abortOnNegative(STREAM(scl.err), id = tbldel(&b->stream->buffers, id, 0));

  /* And processor */
  while (tblget(&b->processor->buffers, id = tblgetnexti(&b->processor->buffers, id), &b2))
    if (b == b2)
      abortOnNegative(STREAM(scl.err), id = tbldel(&b->processor->buffers, id, 0));

  /* ... And global buffer list */
  abortOnNegative(STREAM(scl.err), tbldel(&sc_bufs, b->id, 1));

  scos_mutex_unlock(&v->lock);

  scManCheckDestroyVariable(v);
}

sc_thread_t *scManAddThread(void)
{
  int id;
  sc_thread_t *t;

  id = tbladdnew(&sc_threads, &t);
  t->id = id;
  id = scos_self();
  tbladdhash(&sc_threads, &id, t->id);

  lprintf(STREAM(scl.notice), "New thread %d hashed at %d", t->id, id);

  return t;
}

void scManDestroyThread(void)
{
  int id;
  sc_thread_t *t;

  id = scos_self();
  t = tblget(&sc_threads, tblfindhash(&sc_threads, &id), NULL);

  lprintf(STREAM(scl.notice), "Destroying thread %d at hash %d", t->id, id);

  tbldelhash(&sc_threads, &id);
  tbldel(&sc_threads, t->id, 1);
}

sc_thread_t *scManGetThread(void)
{
  int id;
  sc_thread_t *t;
  char buf[PATH_MAX] = {0};

  id = scos_self();  
  
  t = tblget(&sc_threads, tblfindhash(&sc_threads, &id), NULL);

  if (!t)
  {
    sprintf(buf, "Couldn't find hash %d in threads ", id);

    id = -1;
    while (tblgetnextelement(&sc_threads, &id, &t))
    {
      sprintf(&buf[strlen(buf)], "%d ", t->id);
    }

    sprintf(&buf[strlen(buf)], "\n");
    
#if 0
    lprintf(STREAM(scl.debug), buf);
#endif

    return NULL;
  }

  return t;
}

#if 0
/* 
 * Inserts a dummy operation that replaces the producer of V. This
 * operation can run on a different "processor" than the original producer,
 * such as an external memory. The original producer of V will produce
 * the input to the new operation.
 */
sc_op_t *scManInsertLoop(sc_var_t *v)
{
  sc_var_t *new;
  sc_op_t *o;
  sc_instance_t *i, *orig_instance=v->producer->instance;
  sc_arg_t *a;
  int id;

  lprintf(STREAM(scl.info), "Creating a new cycle operation for %s.%d (producing %d.%d)", v->producer->operation->text, v->producer->id, v->id, v->producer->instance->id);
  
  /* Create new variable and instance */
  new = scManAddVariable(v->type, v->size);
  i = scManAddInstance(new, NULL);
  
  /* The old producer is redirected to the new variable and instance */
  new->producer = v->producer;
  new->producer->instance = i;
  v->producer = NULL;

  /* Create new operation */
  abortOnNegative(STREAM(scl.err), id = tbladdnew(&sc_ops, &o));
  abortOnNegative(STREAM(scl.err), tblcreate(&o->arguments, sizeof(sc_arg_t)));
  o->id = id;
  o->type = CYCLE;
  o->line = -1;
  o->parent = new->producer->operation->parent;
  o->parent_condition = new->producer->operation->parent_condition;
  o->text = "cycle";
  o->format = "iI";
  
  /* Create input, which consumes the new variable */
  abortOnNegative(STREAM(scl.err), id = tbladdnew(&o->arguments, &a));
  a->id = id;
  a->intent = SC_INTENT_IN;
  a->operation = o;
  a->instance = i;
  abortOnNegative(STREAM(scl.err), tbladd(&new->consumers, a));

  /* Create output, which produces the original variable */
  abortOnNegative(STREAM(scl.err), id = tbladdnew(&o->arguments, &a));
  a->id = id;
  a->intent = SC_INTENT_OUT;
  a->operation = o;
  a->buffer = SC_FRAME_BUFFER;
  a->instance = orig_instance;
  v->producer = a;

  i->finalized = 1;

  /* Maybe dispatch? */
  
  return o;
}
#else
/* 
 * Creates a cycle operation as a consumer of the (unfinalized) variable v,
 * and finalizes all instances of v. New instances will be created (at the
 * same addresses as the old ones), pointing to the output of the cycle
 * operation.
 * If such cycles are added for all unfinalized variables, all operations
 * are guaranteed to finish, provided the cycle's output buffer is large
 * enough. Note: cycle's output will have NO consumers!
 *
 * The benefit is that operations that are already consuming v can
 * keep doing so locally, and don't have to go through the cycle, and that
 * the cycle's consumers are easier to dispatch, because only the producer
 * has to be taken into account.
 */
sc_op_t *scManInsertLoop(sc_var_t *v)
{
  sc_var_t *new;
  sc_op_t *o;
  sc_instance_t *i, *i2;
  sc_arg_t *a;
  int id=-1;

  if (scManIsVariableFinalized(v))
    return NULL;

  lprintf(STREAM(scl.info),
	  "Creating a new cycle operation for %s.%d (producing %d)",
	  v->producer->operation->text, v->producer->id, v->id);
  
  /* Create new variable */
  new = scManAddVariable(v->type, v->size);

  /* Add instances for all the old variable's instances, BUT 
   * don't put them into the global hashtable yet */
  while (tblgetnextelement(&v->instances, &id, &i))
  {
    abortOnNegative(STREAM(scl.err), tbladdnewi(&new->instances, &i2, id));
    *i2 = *i;
    i2->variable = new;
    abortOnNegative(STREAM(scl.err),
		    tbladdhash(&new->instances, &i2->address, i2->id));
  }

  /* Create new operation */
  abortOnNegative(STREAM(scl.err), id = tbladdnew(&sc_ops, &o));
  abortOnNegative(STREAM(scl.err), tblcreate(&o->arguments, sizeof(sc_arg_t)));
  o->id = id;
  o->type = CYCLE;
  o->line = v->producer->operation->line;
  o->parent = v->producer->operation->parent;
  o->parent_condition = v->producer->operation->parent_condition;
  o->text = "cycle";
  o->format = "iI";
  
  /* Create input, which consumes the old variable */
  abortOnNegative(STREAM(scl.err), id = tbladdnew(&o->arguments, &a));
  a->id = id;
  a->intent = SC_INTENT_IN;
  a->operation = o;
  a->instance = v->producer->instance;
  abortOnNegative(STREAM(scl.err), tbladd(&v->consumers, a));

  /* Create output, which produces the new variable */
  abortOnNegative(STREAM(scl.err), id = tbladdnew(&o->arguments, &a));
  a->id = id;
  a->intent = SC_INTENT_OUT;
  a->operation = o;
  a->buffer = SC_FRAME_BUFFER;
  a->instance = tblget(&new->instances, v->producer->instance->id, NULL);
  new->producer = a;

  /* Nice bonus: we don't copy v's originrefs! That means we're independent,
   * which happens to be true! */

  /* Finalize all of v's instances, and add our new ones instead */
  while (tblgetnextelement(&new->instances, &id, &i))
    if (!i->finalized)
    {
      _finalize(i->address);
      abortOnNegative(STREAM(scl.err),
		      tbladdhash(&sc_vars, &i->address, new->id));
    }

  /* Maybe dispatch? */
  
  return o;
}
#endif

/*
int scManGetDelay(int op)
{
  int delay = 0;
  scs_operation_t *o;
  scs_target_operation_t *to;

  if (tblget(&scs_operations, op, &o))
  {
    int id=-1;

    while (tblgetnextelement(&o->targets, &id, &to))
      delay = MAX(delay, to->delay);
  }  

  return delay;
}
*/
